package com.suxg.hbase.utils;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.BufferedMutator;
import org.apache.hadoop.hbase.client.BufferedMutatorParams;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.mortbay.log.Log;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.ObjectUtils;

/**
 * Hbase 工具转换
 */
public class HBaseUtilsFilters2 {
	private static final Logger logger = LoggerFactory.getLogger(HBaseUtilsFilters2.class);

	private static Connection connection;

	static {
		// localhost
		Configuration configuration = HBaseConfiguration.create();
		configuration.set("hbase.zookeeper.property.clientPort", "2181");
		// 如果是集群 则主机名用逗号分隔
		configuration.set("hbase.zookeeper.quorum", "localhost:2181");
		configuration.set("hbase.nodeParent", "localhost:2181");
		// aliyun
//		Configuration configuration = HBaseConfiguration.create();
//		configuration.set("hbase.zookeeper.property.clientPort", "2181");
//		// 如果是集群 则主机名用逗号分隔
//		configuration.set("hbase.zookeeper.quorum", "101.200.212.148:2181");
//		configuration.set("hbase.nodeParent", "101.200.212.148:2181");

		try {
			connection = ConnectionFactory.createConnection(configuration);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 判断表是否存在
	 * 
	 * @param tableName
	 * @return
	 * @throws Exception
	 */
	public static boolean tableExists(String tableName) throws Exception {
		HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
		if (admin.tableExists(tableName)) {
			return true;
		} else {
			return false;
		}
	}

	/**
	 * 创建 HBase 表
	 * 
	 * @param tableName      表名
	 * @param columnFamilies 列族的数组
	 */
	public static boolean createTable(String tableName, List<String> columnFamilies) {
		try {
			HBaseAdmin admin = (HBaseAdmin) connection.getAdmin();
			if (admin.tableExists(tableName)) {
				return false;
			}
			HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
			columnFamilies.forEach(columnFamily -> {
				HColumnDescriptor columnDescriptor = new HColumnDescriptor(columnFamily);
				columnDescriptor.setMaxVersions(1);
				tableDescriptor.addFamily(columnDescriptor);
			});
			admin.createTable(tableDescriptor);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return true;
	}

	// *******************Hbase put 部分*************************
	/**
	 * 插入数据
	 *
	 * @param tableName        表名
	 * @param rowKey           唯一标识
	 * @param columnFamilyName 列族名
	 * @param pairList         列标识和值的集合
	 */
	public static boolean putRow(String tableName, String rowKey, String columnFamilyName,
			List<Pair<String, String>> pairList) {
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Put put = new Put(Bytes.toBytes(rowKey));
//            pairList.forEach(pair -> put.addColumn(Bytes.toBytes(columnFamilyName), Bytes.toBytes(pair.getKey()), Bytes.toBytes(pair.getValue())));

			for (Pair<String, String> pair : pairList) {
				put.addColumn(Bytes.toBytes(columnFamilyName), pair.getFirst().getBytes(), pair.getSecond().getBytes());
			}
			table.put(put);
			table.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return true;

	}
	/**
	 * 批量插入
	 * 
	 * @param tablename
	 * @param putList
	 * @return 
	 * @return
	 */
	public static boolean putBath(String tablename, List<Put> putList) {
		BufferedMutator mutator = null;
		TableName tName = TableName.valueOf(tablename);
		BufferedMutatorParams params = new BufferedMutatorParams(tName);
		params.writeBufferSize(5 * 1024 * 1024); // 可以自己设定阈值 5M 达到5M则提交一次
		try {
			mutator = connection.getBufferedMutator(params);
			mutator.mutate(putList); // 数据量达到5M时会自动提交一次
			mutator.flush(); // 手动提交一次
			return true;
		} catch (Exception e) {
			e.printStackTrace();
			return false;
		} finally {
			try {
				if (mutator != null)
					mutator.close(); // 提交一次
			} catch (Exception e) {
				e.printStackTrace();
			}
		}

	}

	/**
	 * 存入list对象，模拟开发人员无法解读list对象
	 * 
	 * @param tableName
	 * @param rowKey
	 * @param columnFamilyName
	 * @param pairList
	 * @return
	 */
	public static boolean putRowObject(String tableName, String rowKey, String columnFamilyName,
			List<Pair<String, String>> pairList) {
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Put put = new Put(Bytes.toBytes(rowKey));
//            pairList.forEach(pair -> put.addColumn(Bytes.toBytes(columnFamilyName), Bytes.toBytes(pair.getKey()), Bytes.toBytes(pair.getValue())));

			for (Pair<String, String> pair : pairList) {
				put.addColumn(Bytes.toBytes(columnFamilyName), pair.getFirst().getBytes(), pair.getSecond().getBytes());
			}
			table.put(put);
			table.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
		return true;

	}

	// *******************Hbase get 部分*************************

	/**
	 * 根据 rowKey 获取指定行的数据(Hbase shell: get)
	 * 
	 * @param tableName
	 * @param rowKey
	 * @param famliy
	 * @param object
	 * @return
	 * @return
	 * @throws Exception
	 */
	public static Object getRow(String tableName, String rowKey, String famliy, Object object) throws Exception {
		Table table = connection.getTable(TableName.valueOf(tableName));
		ConverHbaseMapToObject(rowKey, famliy, object, table);
		return object;
	}

	/**
	 * 获取指定行指定列 (cell) 的最新版本的数据
	 *
	 * @param tableName    表名
	 * @param rowKey       唯一标识
	 * @param columnFamily 列族
	 * @param qualifier    列标识
	 */
	public static String getCell(String tableName, String rowKey, String columnFamily, String qualifier) {
		try {
			Table table = connection.getTable(TableName.valueOf(tableName));
			Get get = new Get(Bytes.toBytes(rowKey));
			if (!get.isCheckExistenceOnly()) {
				get.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier));
				Result result = table.get(get);
				byte[] resultValue = result.getValue(Bytes.toBytes(columnFamily), Bytes.toBytes(qualifier));
				return Bytes.toString(resultValue);
			} else {
				return null;
			}

		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}

	/**
	 * Get查询后转换对象
	 * 
	 * @param rowKey
	 * @param family
	 * @param object
	 * @param table
	 * @throws IOException
	 * @throws IllegalAccessException
	 * @throws InvocationTargetException
	 */
	public static <T> void ConverHbaseMapToObject(String rowKey, String family, Object object, Table table)
			throws IOException, IllegalAccessException, InvocationTargetException {
		Get get = new Get(Bytes.toBytes(rowKey));
		Result resultCell = table.get(get);
		Map<byte[], byte[]> familyMap = resultCell.getFamilyMap(Bytes.toBytes(family));
		Map<String, String> maps = new HashMap<>();
		for (Map.Entry<byte[], byte[]> entry : familyMap.entrySet()) {
			Log.info((Bytes.toString(entry.getKey()) + "=====>" + Bytes.toString((entry.getValue()))));
			if (ObjectUtils.isEmpty(entry.getValue())) {
				maps.put(Bytes.toString(entry.getKey()), "");
			} else {
				maps.put(Bytes.toString(entry.getKey()), Bytes.toString(entry.getValue()));
			}
		}
		BeanUtils.populate(object, maps);
	}

	// *******************Hbase Scanner
	// 部分，必须在scan中设置startRowKey,endRowKey*************************

	/**
	 * 检索表中指定数据
	 *
	 * @param tableName   表名
	 * @param startRowKey 起始 RowKey
	 * @param endRowKey   终止 RowKey
	 * @param filterList  过滤器
	 * @param family      指定列族
	 */

	public static ResultScanner getScanner(Table table, String startRowKey, String endRowKey, FilterList filterList,
			byte[] family) {
		try {
			Scan scan = new Scan();
			scan.setStartRow(Bytes.toBytes(startRowKey));
			scan.setStopRow(Bytes.toBytes(endRowKey));
			scan.setFilter(filterList);
			if (!ObjectUtils.isEmpty(family)) {
				scan.addFamily(family);
			}
			return table.getScanner(scan);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return null;
	}

	/**
	 * 对Hbase的ResultScanner结果做转换
	 * 
	 * @param scanner Hbase 查出的结果
	 * @param object  转换为实体对象
	 * @return 返回List中的实体对象
	 */
	public static <T> List<T> convertScannerToList(ResultScanner scanner, Class<T> object) {
		Map<String, String> maps = new HashMap<>();
		List<T> rs = new ArrayList<>();
		T nweno = null;
		try {
			for (Result result : scanner) {
				// 获取rowkey
				String rowKey = Bytes.toString(result.getRow());
				maps.put("rowKey", rowKey);
				System.out.println("rowkey=====" + rowKey);

//				rs.add(mapper.mapRow(result, rowNum++, null));
				Iterator<KeyValue> it = result.list().iterator();
				while (it.hasNext()) {
					KeyValue kv = it.next();
					// 获取ROWKEY 做为debug使用
//				int rowlength = Bytes.toShort(kv.getBuffer(), kv.getOffset()+KeyValue.ROW_OFFSET); 	
//				String rowKey = Bytes.toStringBinary(kv.getBuffer(), kv.getOffset()+KeyValue.ROW_OFFSET + Bytes.SIZEOF_SHORT, rowlength);
					// Log.info(rowKey+"@@@@@@@@key@@@@key=" + Bytes.toString(kv.getQualifier()) +
					// ", value="+ Bytes.toString(kv.getValue()));
//					CellUtil.getQualifierArray
					Log.info("@@@@@@@@key@@@@key=" + Bytes.toString(kv.getQualifier()) + ", value="
							+ Bytes.toString(kv.getValue()));
					if (ObjectUtils.isEmpty(kv.getValue())) {
						maps.put(Bytes.toString(kv.getQualifier()), "");
					} else {
						maps.put(Bytes.toString(kv.getQualifier()), Bytes.toString(kv.getValue()));
					}
					nweno = object.newInstance();
				}
				BeanUtils.populate(nweno, maps);
				rs.add(nweno);

			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			scanner.close();
		}
		return rs;
	}

	/**
	 * 对Hbase的ResultScanner结果做转换
	 * 
	 * @param scanner Hbase 查出的结果
	 * @param object  转换为实体对象
	 * @return 返回List中的实体对象
	 */
	public static <T> List<T> convertScannerToListMuilts(ResultScanner scanner, Map<String, Class> map,
			Class tableMode) {

		List<T> rs = new ArrayList<>();
		T nweno = null;
		try {
			for (Result result : scanner) {
				Map<String, Object> mapsj = new HashMap<>();
				// 获取rowkey
				String rowKey = Bytes.toString(result.getRow());
//				rs.add(mapper.mapRow(result, rowNum++, null));
				Iterator<KeyValue> it = result.list().iterator();
				Map<String, String> maps = new HashMap<>();
				String family = null;
				while (it.hasNext()) {

					KeyValue kv = it.next();
					// 获取ROWKEY 做为debug使用
//				int rowlength = Bytes.toShort(kv.getBuffer(), kv.getOffset()+KeyValue.ROW_OFFSET); 	
//				String rowKey = Bytes.toStringBinary(kv.getBuffer(), kv.getOffset()+KeyValue.ROW_OFFSET + Bytes.SIZEOF_SHORT, rowlength);
					// Log.info(rowKey+"@@@@@@@@key@@@@key=" + Bytes.toString(kv.getQualifier()) +
					// ", value="+ Bytes.toString(kv.getValue()));
//					CellUtil.getQualifierArray
					family = Bytes.toString(kv.getFamily());
					System.out.println("#####family=" + family);
					Log.info("@@@@@@@@key@@@@key=" + Bytes.toString(kv.getQualifier()) + ", value="
							+ Bytes.toString(kv.getValue()));
					if (ObjectUtils.isEmpty(kv.getValue())) {
						maps.put(Bytes.toString(kv.getQualifier()), "");
					} else {
						maps.put(Bytes.toString(kv.getQualifier()), Bytes.toString(kv.getValue()));
					}

					maps.put("rowKey", rowKey);
					System.out.println("rowkey=====" + rowKey);

					for (Iterator iterator = ((Iterable<Entry<String, Class>>) map).iterator(); iterator.hasNext();) {
						Entry<String, ?> entry = (Entry<String, ?>) iterator.next();
						String key = entry.getKey();
						T value = (T) entry.getValue();
						if (family.equals(key)) {
							nweno = (T) ((Class) value).newInstance();
							BeanUtils.populate(nweno, maps);
						}
					}

					mapsj.put("rowKey", rowKey);
					mapsj.put(family + "Model", nweno);

					T table = (T) tableMode.newInstance();
					BeanUtils.populate(table, mapsj);
					rs.add((T) table);

				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			scanner.close();
		}
		return rs;
	}

	/**
	 * 对Hbase的ResultScanner结果做转换
	 * 
	 * @param scanner Hbase 查出的结果
	 * @param c,      转换为实体对象
	 * @return 返回List中的实体对象
	 */
	public static <T> List<T> convertScannerCellToListMuilts(ResultScanner scanner, Map<String, Class> map,
			Class<T> tableMode) {
		Map<String, String> maps = new HashMap<>();
		Map tmepMap = new HashMap();
		List<T> rs = new ArrayList<>();
		T c = null;
		try {
			for (Result result : scanner) {
				Cell[] cells = result.rawCells();
				String rowKey1 = Bytes.toString(result.getRow());
				T nweno = null;
				Map<String, Object> mapsj = new HashMap<>();
				String family1 = "";
				for (Cell cell : cells) {
					String rowKey = Bytes.toString(CellUtil.cloneRow(cell));
					String family = Bytes.toString(CellUtil.cloneFamily(cell));
					String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
					String qualifierValue = Bytes.toString(CellUtil.cloneValue(cell));
					maps.put("rowKey", rowKey);
//				TestColumnMuitsModel tcmm=new TestColumnMuitsModel();
//				tcmm.setRowKey(rowKey);
					if (ObjectUtils.isEmpty((CellUtil.cloneValue(cell)))) {
						maps.put(Bytes.toString(CellUtil.cloneQualifier(cell)), "");
					} else {
						maps.put(qualifier, qualifierValue);
					}
//			System.out.println("rowkey="+rowKey+"-----family="+family+"----qualifier="+qualifier+"----qualifierValue="+qualifierValue);
					System.out.println("family=" + family);
					if (!tmepMap.containsKey(family + qualifier)) {
						tmepMap.put(family + ":" + qualifier, qualifierValue);
						family1 = family;
						continue;
					}

					tmepMap.put(family1, maps);
					nweno = (T) map.get(family);

					c = ((Class<T>) nweno).newInstance();
//					System.out.println(c.getClass().getName()+"#####"+c.getClass()+"#####"+c.getClass().getSimpleName());
					BeanUtils.populate(c, (Map) tmepMap.get(family));
					mapsj.put("rowKey", rowKey1);
					mapsj.put(toLowerCaseFirstOne(c.getClass().getSimpleName()), c);
//					Field[] fields=	tableMode.getClass().getDeclaredFields();
//					mapsj.put(toLowerCaseFirstOne(c.getClass().getSimpleName()), c);
					T tcmm = (T) tableMode.newInstance();
					BeanUtils.populate(tcmm, mapsj);
					rs.add((T) tcmm);

				}

			}

		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			scanner.close();
		}
		return rs;
	}

	/**
	 * 检索全表
	 * 
	 * @param <T>
	 *
	 * @param tableName 表名
	 */
	public static <T> List<T> ScannerList(String tableName, Class<T> object) {

		Table table = null;
		List<T> rs = new ArrayList<>();
		ResultScanner scanner = null;
		try {
			table = connection.getTable(TableName.valueOf(tableName));
			Scan scan = new Scan();
			// 进行全表扫描加开始，结束行，优化查询效率
//            scan.setStartRow(null);
//            scan.setStopRow(null);

//			Get get = new Get(Bytes.toBytes("rowKey1"));
//			Result resultCell = table.get(get);
//			Map<byte[], byte[]> familyMap = resultCell.getFamilyMap(Bytes.toBytes("student"));
//			for (Map.Entry<byte[], byte[]> entry : familyMap.entrySet()) {
//				System.out.println(Bytes.toString(entry.getKey()));
//			}

			int caching = scan.getCaching();
			// 如果caching未设置(默认是1)，将默认配置成5000
			if (caching == 1) {
				scan.setCaching(5000);
			}
			scanner = table.getScanner(scan);
//			int rowNum = 0;
			Map<String, String> maps = new HashMap<>();
			T nweno = null;
			for (Result result : scanner) {
//				rs.add(mapper.mapRow(result, rowNum++, null));
				Iterator<KeyValue> it = result.list().iterator();
				while (it.hasNext()) {
					KeyValue kv = it.next();
//					CellUtil.getQualifierArray
					System.out.println("@@@@@@@@key@@@@key=" + Bytes.toString(kv.getQualifier()) + ", value="
							+ Bytes.toString(kv.getValue()));
					if (ObjectUtils.isEmpty(kv.getValue())) {
						maps.put(Bytes.toString(kv.getQualifier()), "");
					} else {
						maps.put(Bytes.toString(kv.getQualifier()), Bytes.toString(kv.getValue()));
					}
					nweno = object.newInstance();
				}
				BeanUtils.populate(nweno, maps);
				rs.add(nweno);

			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			scanner.close();
		}
		return rs;
	}

	/**
	 * 首字母转小写
	 * 
	 * @param s
	 * @return
	 */
	public static String toLowerCaseFirstOne(String s) {
		if (Character.isLowerCase(s.charAt(0)))
			return s;
		else
			return (new StringBuilder()).append(Character.toLowerCase(s.charAt(0))).append(s.substring(1)).toString();
	}
}
